package gazzera.node.core.peer;

import gazzera.node.core.*;
import gazzera.node.core.messages.*;
import gazzera.utils.*;

import java.io.*;
import java.util.concurrent.*;

import javax.net.ssl.*;


public class Peer implements Node
{
	private SSLSocket socket;
	private String hash;
	protected BlockingQueue<Object> outbox = new ArrayBlockingQueue<Object>(100);

	private DataInputStream dis;
	private DataOutputStream dos;
	
	boolean isClosed = false;

	public Peer(SSLSocket socket) throws Exception
	{
		super();
		this.socket = socket;
		this.hash = SSLUtils.getSha1Cert(socket.getSession().getPeerCertificates()[0]).toLowerCase();
		this.dis = new DataInputStream(socket.getInputStream());
		this.dos = new DataOutputStream(socket.getOutputStream());
	}

	public void start( ICore core)
	{
		ThreadRunner.start( new PeerReader(this, core), "Peer Reader "+hash, true);
		ThreadRunner.start( new PeerWriter(this, core), "Peer Writer "+hash, true);
	}
	
	@Override
	public boolean equals(Object obj)
	{
		if (obj instanceof Peer)
		{
			return this.hash.equals(((Peer) obj).hash);
		}
		return false;
	}

	@Override
	public String toString()
	{
		return "PEER:" + hash;
	}

	/**
	 * Peer reader is the only thread allowed to recv
	 * @return
	 * @throws IOException
	 */
	protected Message recv() throws IOException
	{
		
		Message o;
		try
		{
			o = Message.readOk(dis);
		}
		catch (Exception e)
		{
			throw new IOException("Errore receiving object "+e.getMessage(), e);
		}
	    if(o==null) throw new IOException("End of socket stream");
	    return o;
	}
	
	protected synchronized void send(Message m) throws IOException
	{
		m.writeOk(dos);
		dos.flush();
	}
	
	/* (non-Javadoc)
	 * @see net.sourceforge.gazzera2.core.peer.Node#close()
	 */
	public synchronized void close()
	{
		if(!isClosed)
		{
			isClosed = true;
			try
			{
				socket.close();
			}
			catch (IOException e)
			{
				
			}
			// poison the writer thread to be sure it will terminate
			outbox.clear();
			outbox.offer(new Poison());
		}
	}
	
	/* (non-Javadoc)
	 * @see net.sourceforge.gazzera2.core.peer.Node#enqueue(net.sourceforge.gazzera2.core.Message)
	 */
	public void enqueue(Message m)
	{
	    if(isClosed) return;
		try
		{
			outbox.put(m);
		}
		catch (Exception e)
		{
			throw new RuntimeException(e);
		}
	}
	/* (non-Javadoc)
	 * @see net.sourceforge.gazzera2.core.peer.Node#enqueue(net.sourceforge.gazzera2.core.DataPumpSource)
	 */
	public void enqueue(DataPumpSource src)
	{
	    if(isClosed) return;
		try
		{
			outbox.put(src);
		}
		catch (Exception e)
		{
			throw new RuntimeException(e);
		}
	}
	/* (non-Javadoc)
	 * @see net.sourceforge.gazzera2.core.peer.Node#getHash()
	 */
	public String getHash()
	{
		return hash;
	}
}
